package com.sure0000.basicKafka.job;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.io.*;
import java.math.BigInteger;
import java.util.*;

/**
 * @author xuyouchang
 * Created on 2021/5/21.
 */
public class SuperJsonToJson {
    private static final String BOOTSTRAP_SERVER = "172.17.1.104:9092,172.17.1.105:9092,172.17.1.106:9092";
    private static final String TOPIC = "ods.origin.snapshot.vehicle.info.in";
    private static final Integer PARTITIONS = 3;
    private static final Short REPLICAS = 1;
    private static final String CONSUMER_GROUP = "test4";

    private static Properties getProps(String bootstrap, String group) {
        Properties props=new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,bootstrap);
        props.put(ConsumerConfig.GROUP_ID_CONFIG,group);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class);
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");
        return props;
    }


    public static String trimBothEndsChars(String srcStr, String splitter) {
        String regex = "^" + splitter + "*|" + splitter + "*$";
        return srcStr.replaceAll(regex, "");
    }
    public static void main(String[] args) throws Exception {
        String pathname = "D:\\ods.afterVehicle.vehicle.info.txt";
        File filename = new File(pathname);
        InputStreamReader reader = new InputStreamReader(
                new FileInputStream(filename));
        BufferedReader br = new BufferedReader(reader);

        File writename = new File("D:\\output.txt");
        writename.createNewFile(); // 创建新文件
        BufferedWriter out = new BufferedWriter(new FileWriter(writename,true));

        List<String> times = new ArrayList<>();
        times.add("pass_time");
        times.add("receive_time");
        times.add("insert_time");

        String line = "";
        line = br.readLine();
        StringBuffer jsonFile = new StringBuffer();
        int count = 0;
        while (line != null) {
            count++;
            line = br.readLine();
            line = line.substring(1,line.length()-1);

            String[] strArray = line.split(", ");
            Map<String, Object> changeMap = new HashMap<>();
            for (String str : strArray) {
                String[] spilts;
                if (str.contains("image_json")) {
                    spilts = new String[]{"image_json", str.replace("image_json=", "")};
                }else {
                    spilts = str.split("=");
                }
                if (spilts.length < 2) {
                    changeMap.put(spilts[0], null);
                }else {
                    if (times.contains(spilts[0])) {
                        changeMap.put(spilts[0], Long.valueOf(spilts[1]));
                    }
//                    else if(JSONObject.isValidArray(spilts[1])) {
//                        changeMap.put(spilts[0], JSONArray.parse(spilts[1]));
//                    }else if(JSONObject.isValidObject(spilts[1])) {
//                        changeMap.put(spilts[0], JSONObject.parse(spilts[1]));
//                    }
                    else {
                        changeMap.put(spilts[0], spilts[1]);
                    }
                }

            }
            JSONObject jsonObject = new JSONObject(changeMap);
            jsonFile.append(jsonObject.toJSONString()).append("\r\n");
            if (count == 1000) {
                out.write(jsonFile.toString());
                out.flush();
                jsonFile = new StringBuffer();
                count = 0;
            }
        }
        out.flush();
        out.close();
    }
}
